package bigdata.hermesfuxi.eagle.etl.jobs;

import bigdata.hermesfuxi.eagle.etl.bean.DataLogBean;
import bigdata.hermesfuxi.eagle.etl.utils.FlinkUtils;
import bigdata.hermesfuxi.eagle.etl.utils.MyKafkaDeserializationSchema;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 直播打赏积分统计作业
 * 
 * 该作业用于实时统计直播打赏相关的各项指标，通过关联维表数据
 * 实现多维度的数据分析和统计。
 * 
 * 主要功能：
 * 1. 实时统计各个主播收到的打赏积分值
 * 2. 统计打赏礼物的数量和受欢迎礼物TopN排行
 * 3. 支持多维度指标统计，数据输出到ClickHouse进行分析
 * 4. 使用广播流关联礼物维表，实现实时数据关联
 * 
 * 技术特点：
 * - 使用Flink广播流关联MySQL维表数据
 * - 支持实时数据流处理和状态管理
 * - 提供侧输出流用于不同维度的统计
 * - 集成Redis缓存用于大盘数据展示
 * 
 * 数据流程：
 * Kafka -> 数据解析 -> 维表关联 -> 多维度统计 -> ClickHouse/Redis
 * 
 * @author hermesfuxi
 * @version 1.0
 * @since 2021
 */
@Slf4j
public class LiveGiftPointStatistics {
    /**
     * 主程序入口
     * 
     * 程序执行流程：
     * 1. 从Kafka消费用户行为数据
     * 2. 过滤出直播打赏事件数据
     * 3. 关联礼物维表获取积分信息
     * 4. 进行多维度统计计算
     * 5. 输出结果到ClickHouse和Redis
     * 
     * @param args 命令行参数，包含配置文件和Kafka配置文件路径
     * @throws Exception 程序执行过程中的异常
     */
    public static void main(String[] args) throws Exception {
        // 设置并行度为1，便于调试和测试
        FlinkUtils.env.setParallelism(1);
        
        // 从Kafka消费数据，获取包含topic、partition、offset信息的元组
        DataStream<Tuple2<String, String>> kafkaSourceV2 = FlinkUtils.getKafkaSourceV2(args, MyKafkaDeserializationSchema.class);

        // 数据解析和过滤：将Kafka消息解析为DataLogBean，并过滤出直播打赏事件
        SingleOutputStreamOperator<DataLogBean> beanStream = kafkaSourceV2.process(new ProcessFunction<Tuple2<String, String>, DataLogBean>() {
            @Override
            public void processElement(Tuple2<String, String> value, Context ctx, Collector<DataLogBean> out) throws Exception {
                try {
                    // 解析JSON数据为DataLogBean对象
                    DataLogBean dataLogBean = JSON.parseObject(value.f1, DataLogBean.class);
                    if (dataLogBean != null) {
                        // 设置全局ID
                        dataLogBean.setGid(value.f0);
                        @SuppressWarnings("unchecked")
                        Map<String, Object> map = dataLogBean.getProperties();
                        
                        // 过滤出直播打赏事件，并提取主播ID和礼物ID
                        if ("liveReward".equals(dataLogBean.getEventId()) && 
                            map != null && 
                            map.get("anchor_id") != null && 
                            map.get("gift_id") != null) {
                            
                            dataLogBean.setAnchorId(map.get("anchor_id").toString());
                            dataLogBean.setGiftId(Integer.parseInt(map.get("gift_id").toString()));
                            out.collect(dataLogBean);
                        }
                    }
                } catch (Exception e) {
                    log.error("解析数据异常: {}", e.getMessage(), e);
                }
            }
        });
        beanStream.print();

        // 创建MySQL数据源，用于获取礼物维表数据
        DataStreamSource<Tuple4<Integer, String, Integer, Integer>> liveGiftMap = FlinkUtils.env.addSource(new MySQLSource());
        
        // 创建广播状态描述符，用于存储礼物维表数据
        // 被广播的一般是小表或字典表，这里广播礼物信息表
        MapStateDescriptor<Integer, Tuple2<String, Integer>> mapStateDescriptor = new MapStateDescriptor<>("broadcastState", TypeInformation.of(Integer.class), TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {
        }));
        // 将礼物维表数据广播到所有并行实例
        BroadcastStream<Tuple4<Integer, String, Integer, Integer>> broadcast = liveGiftMap.broadcast(mapStateDescriptor);

        // 定义侧输出流标签，用于输出不同维度的统计结果
        OutputTag<Tuple2<String, Integer>> anchorPointsTag = new OutputTag<Tuple2<String, Integer>>("anchorPoints") {
        };
        OutputTag<Tuple2<String, Integer>> giftCountTag = new OutputTag<Tuple2<String, Integer>>("giftCount") {
        };

        // 关联主数据流和广播流，进行维表关联和统计计算
        SingleOutputStreamOperator<DataLogBean> result = beanStream.connect(broadcast).process(new LiveGiftPointProcess(mapStateDescriptor, anchorPointsTag, giftCountTag, 3));
        result.print("result");
        // 输出到ClickHouse做多维度分析

        // 处理主播积分统计侧输出流
        DataStream<Tuple2<String, Integer>> anchorPointsStream = result.getSideOutput(anchorPointsTag);
//        anchorPointsStream.print("anchorPoints");
        // 按主播ID分组求和，然后按固定key分组进行TopN排序
        SingleOutputStreamOperator<List<Map.Entry<String, Integer>>> anchorPointsResult = anchorPointsStream.keyBy(t -> t.f0).sum(1).keyBy(t -> "key").process(new SumOrderProcess());
        anchorPointsResult.print("anchorPoints");
        // 输出到Redis中，并展示大盘数据

        // 处理礼物数量统计侧输出流
        DataStream<Tuple2<String, Integer>> giftCountStream = result.getSideOutput(giftCountTag);
//        giftCountStream.print("giftCount");
        // 按礼物名称分组求和，然后按固定key分组进行TopN排序
        giftCountStream.keyBy(t->t.f0).sum(1).keyBy(t->"key").process(new SumOrderProcess()).print("giftCount");
        // 输出到Redis中，并展示大盘数据

        // 启动Flink作业
        FlinkUtils.env.execute();
    }

    /**
     * MySQL数据源类
     * 
     * 该类用于从MySQL数据库实时读取礼物维表数据，支持增量更新。
     * 通过定时轮询的方式获取礼物信息的变化，并将数据发送到Flink流中。
     * 
     * 主要功能：
     * 1. 建立MySQL数据库连接
     * 2. 定时查询礼物维表数据变化
     * 3. 支持增量数据获取，避免重复处理
     * 4. 提供容错机制，连接异常时自动重试
     * 
     * 数据格式：Tuple4<礼物ID, 礼物名称, 积分值, 删除标记>
     */
    public static class MySQLSource extends RichSourceFunction<Tuple4<Integer, String, Integer, Integer>> {
        /** 控制数据源运行状态的标志位 */
        private boolean flag = true;
        /** MySQL数据库连接对象 */
        private Connection connection;

        /**
         * 初始化方法，在数据源启动时调用
         * 建立MySQL数据库连接
         * 
         * @param parameters Flink配置参数
         * @throws Exception 连接异常
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            try {
                // 加载MySQL JDBC驱动
                Class.forName("com.mysql.cj.jdbc.Driver");
                // 建立数据库连接
                connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/realtime?characterEncoding=UTF-8&serverTimezone=GMT%2B8&useSSL=false", "root", "123456");
                log.info("MySQL连接成功");
            } catch (Exception e) {
                log.error("MySQL连接失败: {}", e.getMessage(), e);
                throw e;
            }
        }

        /**
         * 数据源主运行方法
         * 
         * 该方法在restoreState方法之后执行，负责定时查询MySQL数据库
         * 获取礼物维表的变化数据，并发送到Flink流中。
         * 
         * 查询策略：
         * 1. 首次查询获取所有未删除的礼物数据
         * 2. 后续查询基于updateTime字段获取增量数据
         * 3. 每10秒查询一次，出错时等待5秒后重试
         * 
         * @param ctx 数据源上下文，用于发送数据到流中
         * @throws Exception 查询过程中的异常
         */
        @Override
        public void run(SourceContext<Tuple4<Integer, String, Integer, Integer>> ctx) throws Exception {
            long updateTime = 0;
            while (flag) {
                try {
                    // 构建SQL查询语句，支持增量查询
                    String sql = "SELECT id, name, points, deleted FROM tb_live_gift where updateTime > ? " + (updateTime == 0? "and deleted = 0": "");
                    // 查询数据库：SELECT * FROM t_category WHERE last_update > 上一次查询时间
                    PreparedStatement prepareStatement = connection.prepareStatement(sql);
                    prepareStatement.setDate(1, new Date(updateTime));
                    updateTime = System.currentTimeMillis();

                    // 执行查询并处理结果
                    ResultSet resultSet = prepareStatement.executeQuery();
                    while (resultSet.next()) {
                        Integer id = resultSet.getInt("id");
                        String name = resultSet.getString("name");
                        Integer points = resultSet.getInt("points");
                        Integer deleted = resultSet.getInt("deleted");
                        // 发送数据到流中，格式：Tuple4<礼物ID, 礼物名称, 积分值, 删除标记>
                        ctx.collect(Tuple4.of(id, name, points, deleted));
                    }
                    resultSet.close();
                    prepareStatement.close();
                    // 每10秒查询一次
                    Thread.sleep(10000);
                } catch (Exception e) {
                    log.error("查询MySQL数据异常: {}", e.getMessage(), e);
                    // 出错后等待5秒再重试
                    Thread.sleep(5000);
                }
            }
        }

        /**
         * 取消数据源运行
         * 设置标志位为false，停止数据查询
         */
        @Override
        public void cancel() {
            flag = false;
        }

        /**
         * 关闭数据源
         * 释放数据库连接资源
         * 
         * @throws Exception 关闭过程中的异常
         */
        @Override
        public void close() throws Exception {
            if (connection != null) {
                connection.close();
            }
        }
    }

    /**
     * 直播打赏积分处理函数
     * 
     * 该类继承自BroadcastProcessFunction，用于处理主数据流和广播流的关联。
     * 主要功能是关联用户打赏事件和礼物维表数据，计算积分并输出统计结果。
     * 
     * 处理流程：
     * 1. 接收用户打赏事件数据
     * 2. 从广播状态中获取礼物信息
     * 3. 计算积分并设置到数据中
     * 4. 输出到主流和侧输出流进行统计
     */
    private static class LiveGiftPointProcess extends BroadcastProcessFunction<DataLogBean, Tuple4<Integer, String, Integer, Integer>, DataLogBean> {
        /** 广播状态描述符，用于存储礼物维表数据 */
        MapStateDescriptor<Integer, Tuple2<String, Integer>> mapStateDescriptor;

        /** 主播积分统计侧输出流标签 */
        private OutputTag<Tuple2<String, Integer>> anchorPointsTag;
        /** 礼物数量统计侧输出流标签 */
        private OutputTag<Tuple2<String, Integer>> giftCountTag;

        /**
         * 构造函数
         * 
         * @param mapStateDescriptor 广播状态描述符
         * @param anchorPointsTag 主播积分统计侧输出流标签
         * @param giftCountTag 礼物数量统计侧输出流标签
         * @param topN TopN参数（当前未使用）
         */
        public LiveGiftPointProcess(MapStateDescriptor<Integer, Tuple2<String, Integer>> mapStateDescriptor, OutputTag<Tuple2<String, Integer>> anchorPointsTag, OutputTag<Tuple2<String, Integer>> giftCountTag, Integer topN) {
            this.mapStateDescriptor = mapStateDescriptor;
            this.anchorPointsTag = anchorPointsTag;
            this.giftCountTag = giftCountTag;
        }

        /**
         * 处理主数据流中的元素（用户打赏事件）
         * 
         * 该方法处理从Kafka消费的用户打赏事件，通过关联广播状态中的礼物维表数据，
         * 获取礼物名称和积分信息，并输出到不同的流中进行统计。
         * 
         * @param bean 用户打赏事件数据
         * @param ctx 只读上下文，用于获取广播状态
         * @param out 主输出流收集器
         * @throws Exception 处理过程中的异常
         */
        @Override
        public void processElement(DataLogBean bean, ReadOnlyContext ctx, Collector<DataLogBean> out) throws Exception {
            // 获取广播状态中的礼物维表数据
            ReadOnlyBroadcastState<Integer, Tuple2<String, Integer>> broadcastMap = ctx.getBroadcastState(mapStateDescriptor);
            Integer giftId = bean.getGiftId();
            String anchorId = bean.getAnchorId();
            
            // 检查数据完整性和礼物ID是否存在于维表中
            if (giftId != null && anchorId != null && broadcastMap != null && broadcastMap.contains(giftId)) {
                // 从广播状态中获取礼物信息：Tuple2<礼物名称, 积分值>
                Tuple2<String, Integer> stringIntegerTuple2 = broadcastMap.get(giftId);
                
                if (stringIntegerTuple2 != null) {
                    // 设置积分值到数据中
                    Integer points = stringIntegerTuple2.f1;
                    bean.setPoints(points);

                    // 设置礼物名称到数据中
                    String giftName = stringIntegerTuple2.f0;
                    bean.setGiftName(giftName);

                    // 输出到主数据流
                    out.collect(bean);
                    
                    // 输出到主播积分统计侧输出流：Tuple2<主播ID, 积分值>
                    ctx.output(anchorPointsTag, Tuple2.of(anchorId, points));
                    
                    // 输出到礼物数量统计侧输出流：Tuple2<礼物名称, 数量>
                    ctx.output(giftCountTag, Tuple2.of(giftName, 1));
                }
            }
        }

        /**
         * 处理广播流中的元素（礼物维表数据）
         * 
         * 该方法处理从MySQL数据源获取的礼物维表数据，根据删除标记
         * 更新或删除广播状态中的礼物信息。
         * 
         * @param value 礼物维表数据：Tuple4<礼物ID, 礼物名称, 积分值, 删除标记>
         * @param ctx 上下文，用于操作广播状态
         * @param out 输出流收集器（此处不使用）
         * @throws Exception 处理过程中的异常
         */
        @Override
        public void processBroadcastElement(Tuple4<Integer, String, Integer, Integer> value, Context ctx, Collector<DataLogBean> out) throws Exception {
            Integer id = value.f0;        // 礼物ID
            Integer deleted = value.f3;   // 删除标记
            
            // 获取广播状态进行更新
            BroadcastState<Integer, Tuple2<String, Integer>> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
            
            if(deleted == 1){
                // 如果删除标记为1，从广播状态中移除该礼物
                broadcastState.remove(id);
            }else {
                // 否则，更新或添加礼物信息到广播状态：Tuple2<礼物名称, 积分值>
                broadcastState.put(id, Tuple2.of(value.f1, value.f2));
            }
        }
    }

    /**
     * 求和排序处理函数
     * 
     * 该类用于对统计数据进行聚合和排序，实现TopN功能。
     * 通过定时器机制，定期输出排序后的统计结果。
     * 
     * 主要功能：
     * 1. 累加统计数据到状态中
     * 2. 使用定时器定期触发排序输出
     * 3. 按数值大小降序排列，实现TopN功能
     * 
     * 使用场景：
     * - 主播积分排行榜
     * - 礼物受欢迎程度排行榜
     */
    public static class SumOrderProcess extends KeyedProcessFunction<String, Tuple2<String, Integer>, List<Map.Entry<String, Integer>>> {
        /** 用于存储聚合数据的值状态 */
        private transient ValueState<Map<String, Integer>> mapValueState;

        /**
         * 初始化方法
         * 创建值状态用于存储聚合数据
         * 
         * @param parameters Flink配置参数
         * @throws Exception 初始化异常
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            // 创建值状态，用于存储Map<String, Integer>类型的数据
            mapValueState = getRuntimeContext().getState(new ValueStateDescriptor<>("listValueState", TypeInformation.of(new TypeHint<Map<String, Integer>>() {
            })));
        }

        /**
         * 处理数据元素
         * 
         * 该方法接收统计数据，累加到状态中，并注册定时器。
         * 定时器会在下一个10秒整点触发，输出排序后的结果。
         * 
         * @param value 统计数据：Tuple2<名称, 数值>
         * @param ctx 上下文，用于注册定时器
         * @param out 输出流收集器
         * @throws Exception 处理过程中的异常
         */
        @Override
        public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<List<Map.Entry<String, Integer>>> out) throws Exception {
            // 获取当前状态中的Map数据
            Map<String, Integer> map = mapValueState.value();
            if(map == null){
                // 如果状态为空，创建新的HashMap
                map = new HashMap<>();
            }
            // 累加数据到Map中（如果key已存在，会覆盖原值）
            map.put(value.f0, value.f1);
            // 更新状态
            mapValueState.update(map);

            // 计算下一个10秒整点的时间戳
            long currentTimeMillis = System.currentTimeMillis();
            long triggerTime = currentTimeMillis - currentTimeMillis % 10000 + 10000;
            // 注册处理时间定时器
            ctx.timerService().registerProcessingTimeTimer(triggerTime);
        }

        /**
         * 定时器触发方法
         * 
         * 当定时器触发时，从状态中获取所有数据，按数值降序排序后输出。
         * 这样可以实现定期的TopN统计输出。
         * 
         * @param timestamp 定时器触发时间戳
         * @param ctx 定时器上下文
         * @param out 输出流收集器
         * @throws Exception 处理过程中的异常
         */
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<List<Map.Entry<String, Integer>>> out) throws Exception {
            // 获取状态中的所有数据
            Map<String, Integer> map = mapValueState.value();
            // 按数值降序排序，生成TopN列表
            List<Map.Entry<String, Integer>> list = map.entrySet().stream()
                    .sorted((a, b) -> b.getValue() - a.getValue())
                    .collect(Collectors.toList());
            // 输出排序后的结果
            out.collect(list);
        }
    }
}